#include "iomanager.h"
#include "macro.h"
#include "log.h"

#include <errno.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <string.h>
#include <unistd.h>

namespace sylar {
  /**
   * @brief: IO协程调度还解决了调度器在idle状态下忙等待导致CPU占用率高的问题。IO协程调度器使用一对管道fd来tickle调度协程
   *         当调度器空闲时，idle协程通过epoll_wait阻塞在管道的读描述符上，等管道的可读事件。添加新任务时，tickle方法写管道
   *         idle协程检测到管道可读后退出，调度器执行调度。
   * */

  static sylar::Logger::ptr g_logger = SYLAR_LOG_NAME("system");

  enum EpollCtlOp {
  };

  static std::ostream& operator<< (std::ostream& os, const EpollCtlOp& op) {
    switch((int)op) {
#define XX(ctl) \
        case ctl: \
            return os << #ctl;
      XX(EPOLL_CTL_ADD);
      XX(EPOLL_CTL_MOD);
      XX(EPOLL_CTL_DEL);
      default:
        return os << (int)op;
    }
#undef XX
  }

  static std::ostream& operator<< (std::ostream& os, EPOLL_EVENTS events) {
    if(!events) {
      return os << "0";
    }
    bool first = true;
#define XX(E) \
    if(events & E) { \
        if(!first) { \
            os << "|"; \
        } \
        os << #E; \
        first = false; \
    }
    XX(EPOLLIN);
    XX(EPOLLPRI);
    XX(EPOLLOUT);
    XX(EPOLLRDNORM);
    XX(EPOLLRDBAND);
    XX(EPOLLWRNORM);
    XX(EPOLLWRBAND);
    XX(EPOLLMSG);
    XX(EPOLLERR);
    XX(EPOLLHUP);
    XX(EPOLLRDHUP);
    XX(EPOLLONESHOT);
    XX(EPOLLET);
#undef XX
    return os;
  }

  IOManager::FdContext::EventContext& IOManager::FdContext::getContext(IOManager::Event event) {
    switch(event) {
      case IOManager::READ:
        return read;
      case IOManager::WRITE:
        return write;
      default:
        SYLAR_ASSERT2(false, "getContext");
    }
  }

  void IOManager::FdContext::resetContext(EventContext& ctx) {
    ctx.scheduler = nullptr;
    ctx.fiber.reset();
    ctx.cb = nullptr;
  }

  void IOManager::FdContext::triggerEvent(IOManager::Event event) {
    SYLAR_ASSERT(events & event);
    /// 当前事件被触发之后就要取消掉
    events = (Event)(events & ~event);
    EventContext& ctx = getContext(event);
    if(ctx.cb) {
      ctx.scheduler->schedule(&ctx.cb);
    } else {
      ctx.scheduler->schedule(&ctx.fiber);
    }
    ctx.scheduler = nullptr;
 }

  IOManager::IOManager(size_t threads, bool use_caller, const std::string& name)
          : Scheduler(threads, use_caller, name) {
    // 能监听的最大文件描述符数量
    m_epfd = epoll_create(5000);
    SYLAR_ASSERT(m_epfd > 0);

    int rt = pipe(m_tickleFds);
    SYLAR_ASSERT(!rt);

    epoll_event event{};
    memset(&event, 0, sizeof(epoll_event));
    /// 设置管道为fd输入、边沿触发
    event.events = EPOLLIN | EPOLLET;
    event.data.fd = m_tickleFds[0];
    /// 获得/设置文件状态标记(F_GETFL或F_SETFL).
    rt = fcntl(m_tickleFds[0], F_SETFL, O_NONBLOCK);
    SYLAR_ASSERT(!rt);
    rt = epoll_ctl(m_epfd, EPOLL_CTL_ADD, m_tickleFds[0], &event);
    SYLAR_ASSERT(!rt);
    /// 以fd为下标存储每个fd的上下文
    contextResize(32);
    start();
  }

  IOManager::~IOManager() {
    stop();
    close(m_epfd);
    close(m_tickleFds[0]);
    close(m_tickleFds[1]);

    for(size_t i = 0; i < m_fdContexts.size(); ++i) {
      if(m_fdContexts[i]) {
        delete m_fdContexts[i];
      }
    }
  }

  void IOManager::contextResize(size_t size) {
    m_fdContexts.resize(size);

    for(size_t i = 0; i < m_fdContexts.size(); ++i) {
      if(!m_fdContexts[i]) {
        m_fdContexts[i] = new FdContext;
        m_fdContexts[i]->fd = i;
      }
    }
  }
  /**
   * @brief 添加事件
   * @details fd描述符发生了event事件时执行cb函数
   * @param[in] fd socket句柄
   * @param[in] event 事件类型
   * @param[in] cb 事件回调函数，如果为空，则默认把当前协程作为回调执行体
   * @return 添加成功返回0,失败返回-1
   */
  int IOManager::addEvent(int fd, Event event, std::function<void()> cb) {
    FdContext* fd_ctx = nullptr;
    RWMutexType::ReadLock lock(m_mutex);
    if((int)m_fdContexts.size() > fd) {
      fd_ctx = m_fdContexts[fd];
      lock.unlock();
    } else {
      RWMutexType::WriteLock lock2(m_mutex);
      /// 怪细节：每次扩大1.5倍可以使之前被放弃的内存有机会被拿出来重用（2倍就不行）
      contextResize(fd * 1.5);
      fd_ctx = m_fdContexts[fd];
      lock.unlock();  // 这里应该先扩容再拿吧
    }
    /// 同一个fd不允许重复添加事件
    FdContext::MutexType::Lock lock2(fd_ctx->mutex);
    if(SYLAR_UNLIKELY(fd_ctx->events & event)) {
      SYLAR_LOG_ERROR(g_logger) << "addEvent assert fd =" << fd
                                << " event =" << (EPOLL_EVENTS)event
                                << " fd_ctx.event =" << (EPOLL_EVENTS)fd_ctx->events;
      SYLAR_ASSERT(!(fd_ctx->events & event));
    }

    int op = fd_ctx->events ? EPOLL_CTL_MOD : EPOLL_CTL_ADD;
    epoll_event epevent{};
    epevent.events = EPOLLET | fd_ctx->events | event;
    epevent.data.ptr = fd_ctx;

    int rt = epoll_ctl(m_epfd, op, fd, &epevent);
    if(rt) {
      SYLAR_LOG_ERROR(g_logger) << "epoll_ctl(" << m_epfd << ", "
                                << (EpollCtlOp)op << ", " << fd << ", " << (EPOLL_EVENTS)epevent.events << "):"
                                << rt << " (" << errno << ") (" << strerror(errno) << ") fd_ctx->events="
                                << (EPOLL_EVENTS)fd_ctx->events;
      return -1;
    }

    ++m_pendingEventCount;
    fd_ctx->events = (Event)(fd_ctx->events | event);
    /// 这里用引用，方便下面修改fd_ctx的上下文，为其加入回调函数
    FdContext::EventContext& event_ctx = fd_ctx->getContext(event);
    SYLAR_ASSERT(!event_ctx.scheduler
                 && !event_ctx.fiber
                 && !event_ctx.cb);
    /// 在事件上下文中保存当前调度器，方便事件被触发时将事件加入执行队列
    event_ctx.scheduler = Scheduler::GetThis();
    if(cb) {
      event_ctx.cb.swap(cb);
    } else {
      /**
       * @attention:     int rt = iom->addEvent(fd, sylar::IOManager::WRITE);
                         if(rt == 0) {
                           sylar::Fiber::YieldToHold();
                   未加回调函数的addEvent()只在Hook中出现过（如上代码），add成功后即调用yield退出，而event_ctx.fiber = Fiber::GetThis()
                   将这次事件的回调设置为当前协橙，即下次事件被触发会切换到yield的地方继续执行，从而实现hook
       **/
      event_ctx.fiber = Fiber::GetThis();
      SYLAR_ASSERT2(event_ctx.fiber->getState() == Fiber::EXEC
      ,"state=" << event_ctx.fiber->getState());
    }
    return 0;
  }
  /**
   * @brief 删除事件
   * @param[in] fd socket句柄
   * @param[in] event 事件类型
   * @attention 不会触发事件
   * @return 是否删除成功
   */
  bool IOManager::delEvent(int fd, Event event) {
    RWMutexType::ReadLock lock(m_mutex);
    if((int)m_fdContexts.size() <= fd) {
      return false;
    }
    FdContext* fd_ctx = m_fdContexts[fd];
    lock.unlock();

    FdContext::MutexType::Lock lock2(fd_ctx->mutex);
    if(SYLAR_UNLIKELY(!(fd_ctx->events & event))) {
      return false;
    }

    Event new_events = (Event)(fd_ctx->events & ~event);
    int op = new_events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
    epoll_event epevent{};
    epevent.events = EPOLLET | new_events;
    epevent.data.ptr = fd_ctx;

    int rt = epoll_ctl(m_epfd, op, fd, &epevent);
    if(rt) {
      SYLAR_LOG_ERROR(g_logger) << "epoll_ctl(" << m_epfd << ", "
                                << (EpollCtlOp)op << ", " << fd << ", " << (EPOLL_EVENTS)epevent.events << "):"
                                << rt << " (" << errno << ") (" << strerror(errno) << ")";
      return false;
    }

    --m_pendingEventCount;
    fd_ctx->events = new_events;
    FdContext::EventContext& event_ctx = fd_ctx->getContext(event);
    fd_ctx->resetContext(event_ctx);
    return true;
  }
  /**
   * @brief 取消事件
   * @param[in] fd socket句柄
   * @param[in] event 事件类型
   * @attention 如果该事件被注册过回调，那就触发一次回调事件
   * @return 是否删除成功
   */
  bool IOManager::cancelEvent(int fd, Event event) {
    RWMutexType::ReadLock lock(m_mutex);
    if((int)m_fdContexts.size() <= fd) {
      return false;
    }
    FdContext* fd_ctx = m_fdContexts[fd];
    lock.unlock();

    FdContext::MutexType::Lock lock2(fd_ctx->mutex);
    if(SYLAR_UNLIKELY(!(fd_ctx->events & event))) {
      return false;
    }

    Event new_events = (Event)(fd_ctx->events & ~event);
    int op = new_events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
    epoll_event epevent;
    epevent.events = EPOLLET | new_events;
    epevent.data.ptr = fd_ctx;

    int rt = epoll_ctl(m_epfd, op, fd, &epevent);
    if(rt) {
      SYLAR_LOG_ERROR(g_logger) << "epoll_ctl(" << m_epfd << ", "
                                << (EpollCtlOp)op << ", " << fd << ", " << (EPOLL_EVENTS)epevent.events << "):"
                                << rt << " (" << errno << ") (" << strerror(errno) << ")";
      return false;
    }
    /// 删除之前触发一次事件
    fd_ctx->triggerEvent(event);
    --m_pendingEventCount;
    return true;
  }
  /**
    * @brief 取消所有事件
    * @details 所有被注册的回调事件在cancel之前都会被执行一次
    * @param[in] fd socket句柄
    * @return 是否删除成功
  */
  bool IOManager::cancelAll(int fd) {
    RWMutexType::ReadLock lock(m_mutex);
    if((int)m_fdContexts.size() <= fd) {
      return false;
    }
    FdContext* fd_ctx = m_fdContexts[fd];
    lock.unlock();

    FdContext::MutexType::Lock lock2(fd_ctx->mutex);
    if(!fd_ctx->events) {
      return false;
    }

    int op = EPOLL_CTL_DEL;
    epoll_event epevent;
    epevent.events = 0;
    epevent.data.ptr = fd_ctx;

    int rt = epoll_ctl(m_epfd, op, fd, &epevent);
    if(rt) {
      SYLAR_LOG_ERROR(g_logger) << "epoll_ctl(" << m_epfd << ", "
                                << (EpollCtlOp)op << ", " << fd << ", " << (EPOLL_EVENTS)epevent.events << "):"
                                << rt << " (" << errno << ") (" << strerror(errno) << ")";
      return false;
    }

    if(fd_ctx->events & READ) {
      fd_ctx->triggerEvent(READ);
      --m_pendingEventCount;
    }
    if(fd_ctx->events & WRITE) {
      fd_ctx->triggerEvent(WRITE);
      --m_pendingEventCount;
    }

    SYLAR_ASSERT(fd_ctx->events == 0);
    return true;
  }

  IOManager* IOManager::GetThis() {
    return dynamic_cast<IOManager*>(Scheduler::GetThis());
  }

  void IOManager::tickle() {
    if(!hasIdleThreads()) {
      return;
    }
    int rt = write(m_tickleFds[1], "T", 1);
    SYLAR_ASSERT(rt == 1);
  }
 /**
  * @brief: 停止调度器顺便获得下一次超时时间
  * @param[in,out]: timeout 返回最近的下一次定时器时间
  * */
  bool IOManager::stopping(uint64_t& timeout) {
    timeout = getNextTimer();
    /// 最近没有定时器需要处理且待处理任务为0且调用了stop()
    return timeout == ~0ull
           && m_pendingEventCount == 0
           && Scheduler::stopping();
  }

  bool IOManager::stopping() {
    uint64_t timeout = 0;
    return stopping(timeout);
  }

  void IOManager::idle() {
    // std::cout << "hello world\n" << std::endl;
    SYLAR_LOG_DEBUG(g_logger) << "idle";
    const uint64_t MAX_EVNETS = 256;
    epoll_event* events = new epoll_event[MAX_EVNETS]();
    /// 用智能指针管理数组
    std::shared_ptr<epoll_event> shared_events(events, [](epoll_event* ptr){
      delete[] ptr;
    });

    while(true) {
      /// 获取下一个定时器的超时时间，顺便判断调度器是否停止
      uint64_t next_timeout = 0;
      if(SYLAR_UNLIKELY(stopping(next_timeout))) {
        SYLAR_LOG_INFO(g_logger) << "name=" << getName()
                                 << " idle stopping exit";
        break;
      }
      /// 阻塞在epoll_wait上，等待事件发生或定时器超时
      int rt = 0;
      do {
        /// 默认超时时间5秒，如果下一个定时器的超时时间大于5秒，仍以5秒来计算超时，
        /// 避免定时器超时时间太长，epoll_wait一直阻塞
        static const int MAX_TIMEOUT = 3000;
        if(next_timeout != ~0ull) {
          next_timeout = (int)next_timeout > MAX_TIMEOUT
                         ? MAX_TIMEOUT : next_timeout;
        } else {
          next_timeout = MAX_TIMEOUT;
        }
        /// epoll_wait和epoll_ctl是线程安全的
        rt = epoll_wait(m_epfd, events, MAX_EVNETS, (int)next_timeout);
        /// 返回-1且errno被设置为EINTR表示陷入阻塞的操作被某种信号中断
        if(rt < 0 && errno == EINTR) {
        } else {
          break;
        }
      } while(true);

      /// 收集所有已超时的定时器，执行回调函数
      std::vector<std::function<void()>> cbs;
      listExpiredCb(cbs);
      if(!cbs.empty()) {
        for(const auto &cb : cbs) {
          schedule(cb);
        }
        cbs.clear();
      }

      for(int i = 0; i < rt; ++i) {
        epoll_event& event = events[i];
        // 如果消息从管道里来，表示调用了tickle，这时候只需要把消息都消化掉，那么下次轮询（run）的时候会自动检查任务。
        if(event.data.fd == m_tickleFds[0]) {
          uint8_t dummy[256];
          while(read(m_tickleFds[0], dummy, sizeof(dummy)) > 0);
          continue;
        }
        /**
         * EPOLLERR: 出错，比如写读端已经关闭的pipe
         * EPOLLHUP: 套接字对端关闭
         * 出现这两种事件，应该同时触发fd的读和写事件，否则有可能出现注册的事件永远执行不到的情况
         */
        FdContext* fd_ctx = (FdContext*)event.data.ptr;
        FdContext::MutexType::Lock lock(fd_ctx->mutex);
        /// 这里表示如果触发的事件是EPOLLERR或EPOLLHUP，则应该检查原本注册fd时的事件中有没有IN和OUT
        /// 然后需要将事件执行一下，不然事件就白注册了
        if(event.events & (EPOLLERR | EPOLLHUP)) {
          event.events |= (EPOLLIN | EPOLLOUT) & fd_ctx->events;
        }
        int real_events = NONE;
        if(event.events & EPOLLIN) {
          real_events |= READ;
        }
        if(event.events & EPOLLOUT) {
          real_events |= WRITE;
        }
        // 被触发的事件不需要读或写
        if((fd_ctx->events & real_events) == NONE) {
          continue;
        }
        // 剔除已经发生的事件，将剩下的事件重新加入epoll_wait，
        // 如果剩下的事件为0，表示这个fd已经不需要关注了，直接从epoll中删除
        int left_events = (fd_ctx->events & ~real_events);
        int op = left_events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
        // 这里好像是每次边沿触发之后都要重新设置边沿触发
        event.events = EPOLLET | left_events;

        int rt2 = epoll_ctl(m_epfd, op, fd_ctx->fd, &event);
        if(rt2) {
          SYLAR_LOG_ERROR(g_logger) << "epoll_ctl(" << m_epfd << ", "
                                    << (EpollCtlOp)op << ", " << fd_ctx->fd << ", " << (EPOLL_EVENTS)event.events << "):"
                                    << rt2 << " (" << errno << ") (" << strerror(errno) << ")";
          continue;
        }

        if(real_events & READ) {
          fd_ctx->triggerEvent(READ);
          --m_pendingEventCount;
        }
        if(real_events & WRITE) {
          fd_ctx->triggerEvent(WRITE);
          --m_pendingEventCount;
        }
      }
      /// 取消智能指针的绑定，防止内存泄漏
      Fiber::ptr cur = Fiber::GetThis();
      auto raw_ptr = cur.get();
      cur.reset();
      /// 被唤醒一次就要回到调度协橙执行任务
      raw_ptr->swapOut();
    } // end while(true)
  }

  void IOManager::onTimerInsertedAtFront() {
    tickle();
  }

}